package com.netty.websocket;

import com.alibaba.fastjson.JSONObject;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

public final class ComsumerChannelGroup extends BaseChannelGroup{
    public ConcurrentHashMap<String, ConcurrentLinkedQueue<Channel>> mapConcurrentLinkedQueue = new ConcurrentHashMap();

    public static ConcurrentHashMap<String,ComsumerChannelGroup> mapComsumerChannelGroup=new ConcurrentHashMap();//记录所有队列的SingleQueue
    public static ComsumerChannelGroup newOrFind(String queue,QueueType queueType) {
        String key=queue+"-"+queueType.name();
        ComsumerChannelGroup comsumerChannelGroup = mapComsumerChannelGroup.get(key);
        if (comsumerChannelGroup == null) {
            comsumerChannelGroup = new ComsumerChannelGroup();
            comsumerChannelGroup.queue=queue;
            mapComsumerChannelGroup.put(key, comsumerChannelGroup);
        }
        return comsumerChannelGroup;
    }

    @Override
    public BaseChannelGroup add(NettySession nettySession) {
        queue=nettySession.getQueue();
        ConcurrentLinkedQueue concurrentLinkedQueue = mapConcurrentLinkedQueue.get(queue);
        if (concurrentLinkedQueue == null) {
            concurrentLinkedQueue = new ConcurrentLinkedQueue();//java 队列容器
            mapConcurrentLinkedQueue.put(queue, concurrentLinkedQueue);
        }
        concurrentLinkedQueue.add(nettySession.channel());
        mapQueueSize.put(queue,concurrentLinkedQueue.size());
        return this;
    }

    @Override
    public BaseChannelGroup remove(NettySession nettySession) {
        queue=nettySession.getQueue();
        ConcurrentLinkedQueue<Channel> concurrentLinkedQueue = mapConcurrentLinkedQueue.get(queue);
        if (concurrentLinkedQueue == null) {
            return this;
        }
        concurrentLinkedQueue.remove(nettySession.channel());
        mapQueueSize.put(queue,concurrentLinkedQueue.size());
        return this;
    }

    @Override
    public boolean writeAndFlush(NettySession nettySession,String recvQueue,JSONObject jsonMsg) {
        if (jsonMsg==null) return false;
        return writeAndFlush(nettySession,recvQueue,new TextWebSocketFrame(jsonMsg.toJSONString()));
    }
    @Override
    public boolean writeAndFlush(NettySession nettySession,String recvQueue,WebSocketFrame message) {
        if (message==null) return false;
        ConcurrentLinkedQueue<Channel> concurrentLinkedQueue = mapConcurrentLinkedQueue.get(recvQueue);
        if (concurrentLinkedQueue == null) {
            return false;
        }

        Channel tempChannel=concurrentLinkedQueue.poll();//获取并删除一个
        if (tempChannel!=null && tempChannel==nettySession.channel()){//假如拿到的是自己，这重新获取一个
            concurrentLinkedQueue.add(tempChannel);      //放入到最后

            tempChannel =concurrentLinkedQueue.poll();   //获取并删除一个
            concurrentLinkedQueue.add(tempChannel);
            if (tempChannel==nettySession.channel()){//假如拿到的还是是自己，则return，说明 concurrentLinkedQueue 中只有一个对象
                return false;
            }
        }else if (tempChannel!=null){
            concurrentLinkedQueue.add(tempChannel);
        }
        if (tempChannel!=null) {
            //System.out.println("队列 queue=" + queue + "  " + WebSocketHandler.getChannelId(tempChannel) + " 单一广播消息：" + textMessage.text());
            ChannelFuture future = tempChannel.writeAndFlush(message);//广播消息.writeAndFlush(textMessage);//广播消息
//            future.addListener(new ChannelFutureListener() {
//                public void operationComplete(ChannelFuture future) {
//
//                }
//            });

            return true;
        }
        return false;
    }
}
